package mya.mes.mdc.mqtt;


import mya.mes.mdc.share.mqttwrite.MqttPublishCallback;
import mya.mes.mdc.share.vo.MdcMqttDefaultTopic;
import mya.mes.mdc.share.vo.MdcOptions;
import mya.mes.mdc.share.vo.mqtt.MqttMsgReceived;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.fusesource.hawtbuf.Buffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.*;

import java.net.URISyntaxException;
import java.util.List;


class FuseMqttClient {

    private static final Logger logger = LogManager.getLogger(FuseMqttClient.class);

    private CallbackConnection connection = null;

    private volatile boolean isConnecting = false;

    private final MQTT mqtt;

    private final MqttReceivedQueue mqttMsgQueue;

    private final MsgReceiver msgReceiver;
    private MdcOptions mdcOptions;

    FuseMqttClient(MdcOptions mdcOptions, MqttReceivedQueue mqttMsgQueue, MsgReceiver msgReceiver) {
        this.mdcOptions = mdcOptions;
        this.mqttMsgQueue = mqttMsgQueue;
        this.msgReceiver = msgReceiver;
        mqtt = new MQTT();
        try {
            mqtt.setHost(mdcOptions.MQTT_IP, mdcOptions.MQTT_PORT);
        } catch (URISyntaxException urs) {
            urs.printStackTrace();
        }
        //or  mqttwrite.setHost("tcp://localhost:1883");
        mqtt.setClientId(mdcOptions.MQTT_CLIENT_ID);
        mqtt.setCleanSession(true);
        mqtt.setKeepAlive((short) 20000);
        mqtt.setReconnectAttemptsMax(-1);
        mqtt.setConnectAttemptsMax(-1);
        mqtt.setVersion("3.1.1");
    }

    public void connect() {
        final CallbackConnection connection = mqtt.callbackConnection();
        connection.listener(new Listener() {
            public void onDisconnected() {
                logger.warn("MQTT onDisconnected");
                System.out.println("MQTT onDisconnected");
                debug("MQTT onDisconnected");
                isConnecting = false;
            }
            public void onConnected() {
                logger.info("MQTT onConnected");
                System.out.println("MQTT onConnected");
                debug("MQTT onDisconnected");
                isConnecting = true;
                try {
                    msgReceiver.start();
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
            public void onPublish(UTF8Buffer topic, Buffer payload, Runnable ack) {
                // You can now process a received message from a topic.
                // Once process execute the ack runnable.
                try {
                    String mqttTopic = topic.toString();
                    String mqttContent = payload.ascii().toString();
                    Long mqttId = TimeIdGenerator.getInstance().getId(System.currentTimeMillis());
                    MqttMsgReceived mqttPayload = new MqttMsgReceived();
                    mqttPayload.mqttMsgId = mqttId;
                    mqttPayload.mqttTopic = mqttTopic;
                    mqttPayload.mqttContent = mqttContent;
                    try {
                        mqttMsgQueue.add(mqttPayload);
                    } catch (Throwable t) {
                        t.printStackTrace();
                    }


                    ack.run();
//                Future<MsgProcessTaskResult> future = msgReceiver.processMsg(mqttPayload);
//                try {
//                    MsgProcessTaskResult result = future.get();
//                    if (!result.success) {
//                        //记录错误
//                    }
//                    ack.run();
//                } catch (Exception e) {
//                    e.printStackTrace();
//                }
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
            public void onFailure(Throwable ex) {
                logger.error("MQTT onFailure 失败！", ex);
                debug("MQTT onFailure 失败！" + ex.getMessage());
                disconnect(); // a connection failure occured.
            }
        });

        connection.connect(new Callback<Void>() {
            public void onFailure(Throwable ex) {
                //   result.failure(value); // If we could not connect to the server.
                logger.error("MQTT 连接失败！", ex);
                debug("MQTT 连接失败！" + ex.getMessage());
            }

            // Once we connect..
            public void onSuccess(Void v) {
                logger.info("MQTT 连接成功！");
                subscribe();
            }
        });


        this.connection = connection;
    }


    public void subscribe() {

        if (msgReceiver == null) {
            return;
        }
        List<String> topicList = mdcOptions.SUBSCRIBE_TOPIC_LIST;

        if (topicList == null) {
            if (mdcOptions.SUBSCRIBE_DEFAULT_TOPIC) {
                topicList = MdcMqttDefaultTopic.topicList;
            } else {
                throw new RuntimeException("没有定义subscribe的topc");
            }
        } else {
            mdcOptions.SUBSCRIBE_DEFAULT_TOPIC = false;
        }
        MqttClientManager.getInstance()
                .getDataConverter().setUseMqttDefaultTopic(mdcOptions.SUBSCRIBE_DEFAULT_TOPIC);
        if (topicList == null || topicList.isEmpty()) {
            return;
        }
        Topic[] topics = new Topic[topicList.size()];
        for (int i = 0; i < topicList.size(); i++) {
            String topicName = topicList.get(i);
            System.out.println("MQTT subscribe topic  :" + topicName);
            topics[i] = new Topic(topicName, QoS.AT_LEAST_ONCE);
        }
        connection.subscribe(topics, new Callback<byte[]>() {
            public void onSuccess(byte[] qoses) {
                // The result of the subcribe request.
                logger.info("MQTT subscribe 成功！");
            }
            public void onFailure(Throwable ex) {
                logger.error("MQTT subscribe 失败！", ex);
                debug("MQTT subscribe 失败！" + ex.getMessage());
                disconnect(); // subscribe failed.
            }
        });
    }

    public void publish(String topic, String content, MqttPublishCallback mqttPublishCallback) {
        // Send a message to a topic
        connection.publish(topic, content.getBytes(), QoS.AT_LEAST_ONCE, false, new Callback<Void>() {
            public void onSuccess(Void v) {
                // the pubish operation completed successfully.

                if (mqttPublishCallback != null) {
                    mqttPublishCallback.onSuccess(v);
                }
            }

            public void onFailure(Throwable ex) {
                logger.error("MQTT publish 失败！", ex);
                debug("MQTT publish 失败！" + ex.getMessage());
                disconnect(); // publish failed.

                if (mqttPublishCallback != null) {
                    mqttPublishCallback.onFailure(ex);
                }
            }
        });
    }

    public void disconnect() {
        // To disconnect..
        connection.disconnect(new Callback<Void>() {
            public void onSuccess(Void v) {
                // called once the connection is disconnected.
            }
            public void onFailure(Throwable value) {
                // Disconnects never fail.
            }
        });
    }

    private void debug(String msg) {
        System.out.println(msg);
    }

    public void shutdown() {
        if (connection == null) {
            return;
        }
        disconnect();
    }


}
